package com.源码分析.processor;

import java.util.stream.IntStream;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.publisher.WorkQueueProcessor;

public class WorkQueueProcessorDemo {

	static final Logger LOGGER = LoggerFactory.getLogger(WorkQueueProcessorDemo.class);

	@Test
    public void testWorkQueueProcessor(){
        WorkQueueProcessor<Integer> workQueueProcessor = WorkQueueProcessor.<Integer>builder()
        		.share(true)
//              .executor(Executors.newSingleThreadExecutor())
        		.build();

        // 能者多劳，有点分布式锁的意思，谁拿到锁谁执行
        // 这里，所有的subscriber的request会累加在一起，然后workQueueProcessor每次只给其中一个subscriber数据，也就是所有下游订阅者下发元素的数量总和与上游Publish发布的元素总量相等
        workQueueProcessor.subscribe(e -> {
            LOGGER.info("flux1 subscriber:{}",e);
            try {
				Thread.sleep(100);
			} catch (InterruptedException e1) {
				e1.printStackTrace();
			}
        });
        workQueueProcessor.subscribe(e -> {
            LOGGER.info("flux2 subscriber:{}",e);
            try {
				Thread.sleep(1000);
			} catch (InterruptedException e1) {
				e1.printStackTrace();
			}
        });

        IntStream.range(1,100)
        .parallel()
        .forEach(e -> {
            workQueueProcessor.onNext(e);
        });

        workQueueProcessor.onComplete();
        workQueueProcessor.blockLast();

        try {
			Thread.sleep(10000000);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
    }

}
